package com.ws.datasync.springTask;

import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSON;
import com.ws.datasync.bean.Actual;
import com.ws.datasync.bean.AlarmDto;
import com.ws.datasync.bean.ErrorActualInfo;
import com.ws.datasync.bean.HealthActual;
import com.ws.datasync.controller.IndexController;
import com.ws.datasync.service.CluterHealthService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;

import static java.util.Comparator.comparing;

@Component
public class ScheduledTasks {
    private static Logger logger = LoggerFactory.getLogger(ScheduledTasks.class);


    private static final String MERGE_KEY = "entity_name,entity_addr";
    private static final Integer RETRY_TIME = 4;

    @Autowired
    private CluterHealthService cluterHealthService;

    @Autowired
    private IndexController indexController;

    @Value("${spring.datasource.secondary.type}")
    public String secondaryDBType;  //源数据库类型

    @Value("${spring.datasource.secondary.name}")
    public String secondaryName;  //源数据库名

    @Value("${spring.datasource.primary.name}")
    public String primaryName;   //本地数据库名

    @Value("${spring.datasource.secondary.jdbc-url}")
    public String secondaryurl;

    @Value("${spring.datasource.primary.jdbc-url}")
    public String primaryurl;

    @Value("${table.list}")
    public String table_list;

    @Value("${alarm.uri}")
    public String alarm_uri;

    @Value("${alarm.apikey}")
    public String alarm_apikey;

    @Value("${alarm.appkey}")
    public String alarm_appkey;

    @Autowired
    @Qualifier("primaryJdbcTemplate")
    protected JdbcTemplate jdbcTemplate1; //本地数据库

    @Autowired
    @Qualifier("secondaryJdbcTemplate")
    protected JdbcTemplate jdbcTemplate2;//源数据库

    // 集群状态告警  每天凌晨12 点执行
    @Scheduled(cron = "0 0 0 * * ?")
    public String cronJob1(){
            //获取报错的节点信息
            List<ErrorActualInfo> result = new ArrayList<>();
            List<HealthActual> healthActuals = cluterHealthService.queryHealthActual();
            for (HealthActual healthActual : healthActuals) {
                for (Actual actual : healthActual.getActuals()) {
                    if (actual.getLastErro() != null && !actual.getLastErro().isEmpty()
                            && actual.getLastErroNo() != null && !actual.getLastErroNo().isEmpty()) {
                        result.add(new ErrorActualInfo(healthActual.getName(), healthActual.getJdbcurl()));
                        break;
                    }
                }
            }
            if (!result.isEmpty()) {
                AlarmDto alarmDto = new AlarmDto();
                alarmDto.setName("数据库集群告警");
                String allNames = result.stream()
                        .map(ErrorActualInfo::getName)
                        .collect(Collectors.joining(", "));
                alarmDto.setEntity_name(allNames);
                String allurl = result.stream()
                        .map(info -> {
                            String prefix = info.getJdbcurl().substring("jdbc:mysql://".length());
                            String ips = prefix.split(":")[0];
                            return ips;
                        })
                        .collect(Collectors.joining(", "));
                alarmDto.setEntity_addr(allurl);
                alarmDto.setOccur_time(System.currentTimeMillis());
                alarmDto.setSeverity("1");
                alarmDto.setDescription("数据库集群节点状态异常");
                alarmDto.setMerge_key(MERGE_KEY);
                alarmDto.setIdentify_key("entity_addr");
                //发送告警
                Boolean res_boolean =  send_alarm(alarmDto);
                if(res_boolean){
                    return "集群状态告警已发送";
                }else{
                    return "集群状态告警调用统一监控平台失败";
                }
            }else{
                logger.info("状态正常，无告警");
                return "集群状态无异常";
            }
    }

    //数据同步告警
    @Scheduled(cron = "0 0 0 * * ?")
    public String cronJob2() {
        // 获取当前时间并向前推一个小时
        LocalDateTime now = LocalDateTime.now();
        LocalDateTime oneHourAgo = now.minusHours(1);
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        String formattedDateTime = oneHourAgo.format(formatter);
        List<Map<String, String>> mapList = screenData(formattedDateTime);
        List<Map<String, String>> errorMapList = mapList.stream()
                .filter(map -> !map.get("sourceTableRows").equals(map.get("targetTableRows")))
                .collect(Collectors.toList());
        if(errorMapList!=null && errorMapList.size()>0){
            AlarmDto alarmDto = new AlarmDto();
            alarmDto.setName("表数据同步告警");
            String allNames = errorMapList.stream()
                    .map(map -> map.get("tableName"))
                    .collect(Collectors.joining(","));
            alarmDto.setEntity_name(allNames);
            // 去除前缀部分
            String prefix1 = primaryurl.substring("jdbc:mysql://".length());
            String prefix2 = secondaryurl.substring("jdbc:mysql://".length());

            // 提取 IP 地址部分
            String primaryip = prefix1.split(":")[0];
            String secondip = prefix2.split(":")[0];
            //数据源地址 从集群节点任取一个,用来派单
            List<HealthActual> healthActuals = cluterHealthService.queryHealthActual();
            String nodeJdbcUrl = healthActuals.get(0).getJdbcurl();
            String nodeIp = nodeJdbcUrl.substring("jdbc:mysql://".length()).split(":")[0];
            alarmDto.setEntity_addr(nodeIp);
            alarmDto.setOccur_time(System.currentTimeMillis());
            alarmDto.setSeverity("1");
            alarmDto.setDescription("表数据同步异常");
            alarmDto.setMerge_key(MERGE_KEY);
            alarmDto.setIdentify_key("entity_addr");
            //发送告警
            Boolean res_boolean =  send_alarm(alarmDto);
            if(res_boolean){
                return "表数据同步告警已发送";
            }else{
                return "表数据同步告警调用统一监控平台失败";
            }
        }else{
            logger.info("状态正常，无告警");
            return "表数据同步无异常";
        }
    }






    public List<Map<String, String>> screenData(String formattedDateTime) {

        List<String> tableList = Arrays.asList(table_list.split(","));

        Map<String, String> sourceMap = new HashMap<String, String>();
        tableList.stream().forEach(name -> {
            if("request_task".equals(name)) {
                sourceMap.put(name, jdbcTemplate1.queryForObject("select count(*) from " + name + " WHERE update_time < ? ", String.class,formattedDateTime));
            }else{
                sourceMap.put(name, jdbcTemplate1.queryForObject("select count(*) from " + name + " WHERE UPDATETIME < ? ", String.class,formattedDateTime));
            }
        });
        Map<String, String> targetMap = new HashMap<String, String>();
        tableList.stream().forEach(name -> {
            if("request_task".equals(name)) {
                targetMap.put(name, jdbcTemplate2.queryForObject("select count(*) from " + name + " WHERE update_time < ?", String.class, formattedDateTime));
            }else{
                targetMap.put(name, jdbcTemplate2.queryForObject("select count(*) from " + name + " WHERE UPDATETIME < ?", String.class, formattedDateTime));
            }
        });

        List<Map<String, String>> mergedList = new ArrayList<>();
        Set<String> set = new HashSet<>(tableList);
        for (String name : set) {
            mergedList.add(new HashMap<String, String>(){{
                put("tableName", name);
                put("sourceTableRows", sourceMap.get(name));
                put("targetTableRows", targetMap.get(name));
                logger.info("tableName:{} sourceTableRows:{} targetTableRows:{}", name,sourceMap.get(name),targetMap.get(name) );
            }});
        }
        mergedList = mergedList.stream().sorted(comparing(m -> m.get("tableName").toString()))
                .collect(Collectors.toList());
        return mergedList;
    }



    public Boolean send_alarm(AlarmDto alarmDto){
        //发送告警
        try {
            Map<String,String> headerMap = new HashMap<>();
            headerMap.put("apikey",alarm_apikey);
            headerMap.put("appkey",alarm_appkey);
            alarmDto.setRequestHeader(headerMap);
            String jsondata = JSON.toJSONString(alarmDto);
            logger.info("入参,data:{}",jsondata);
            String uri = alarm_uri;
            Long startTime = null;
            for (int i = 0; i < RETRY_TIME; i++) {
                try {
                    startTime = new Date().getTime();
                    String res = HttpUtil.createPost(uri)
                            .body(jsondata)
                            .header("Content-Type", "application/json;charset=UTF-8")
                            .timeout(20000)
                            .execute().body();
                    if (res.contains("000000")) {
                        logger.info("告警已发送统一监控平台,res:{}",res);
                        return true;
                    }else{
                        logger.info("调用失败，统一监控返回数据,res:{}",res);
                        return false;
                    }
                } catch (Exception e) {
                    if (i == (RETRY_TIME - 1)) {
                        //重试次数达到上限直接抛出
                        logger.error("alertTrace httpPushData retry " + RETRY_TIME + " time error:{}", e);
                        return false;
                    }
                    Long endTime = new Date().getTime();
                    if ((endTime - startTime) < 1000) {
                        //间隔1s 再重试
                        Thread.sleep(endTime - startTime);
                    }
                    logger.info("alertTrace httpPushData retry times:{} msg:{}", i, e);
                }
            }
        }catch (Exception e){
            logger.error(" failed msg:{}" , e);
        }
        return true;
    }

}
